package org.example.reactor;

import com.sun.org.apache.regexp.internal.RE;
import org.example.utils.LogUtils;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * IO事件的业务处理
 * 引入线程池将IO事件的处理和反应器线程分隔开
 */
public class MultiThreadHandler implements Runnable{

    private Selector selector;

    private SocketChannel sch;

    private static final int RECEIVING = 0, SENDING = 1;

    private int state = RECEIVING;

    private ByteBuffer buffer = ByteBuffer.allocate(1024);

    static ExecutorService executorService = Executors.newFixedThreadPool(4);

    private SelectionKey skey;

    public MultiThreadHandler(Selector selector, SocketChannel sch) {
        try {
            this.selector = selector;
            this.sch = sch;
            // 唤醒线程
            this.selector.wakeup();
            sch.configureBlocking(false);
            skey = sch.register(this.selector, 0);
            skey.attach(this);
            skey.interestOps(SelectionKey.OP_READ);

            this.selector.wakeup();
        } catch (IOException e) {
            LogUtils.info(e.getMessage());
        }
    }

    @Override
    public void run() {
        // 如果轮询事件的速度 > 线程内处理速度
        // 可能线程还没有处理完，轮询到了相同事件，线程又提交了一把
        executorService.execute(new Runnable() {
            @Override
            public void run() {
                MultiThreadHandler.this.asyncRun();
            }
        });
//        asyncRun();
    }

    /**
     * 会被多个线程同时访问，所以加了synchronized进行同步
     */
    public synchronized void asyncRun() {
        try {
            buffer.clear();
            if (state == RECEIVING) {
                int num = -1;
                // 是否有内容的标志位
                boolean flag = false;
                while ((num = sch.read(buffer)) > 0) {
                    byte[] contents = new byte[num];
                    buffer.flip();
                    buffer.get(contents);
                    LogUtils.info(new String(contents));
                    buffer.clear();
                    flag = true;
                }

                // 如果没有读取到内容，不注册事件
                if (flag) {
                    skey.interestOps(SelectionKey.OP_WRITE);
                    state = SENDING;
                }
            }
            else if (state == SENDING) {
                buffer.put("welcome to server".getBytes());
                buffer.flip();
                sch.write(buffer);
                skey.interestOps(SelectionKey.OP_READ);
                state = RECEIVING;
            }
        } catch (IOException e) {
            LogUtils.info(e.getMessage());

            skey.cancel();

            try {
                sch.close();
            } catch (IOException ioe) {
                LogUtils.info(e.getMessage());
            }
        }
    }
}
